package server_agent

import (
	"bufio"
	"bytes"
	"encoding/binary"
	"fmt"
	"io"
	"lark/conf"
	"lark/consts"
	"lark/proto"
	"lark/util"
	"strconv"
	"sync/atomic"
	"time"

	gproto "github.com/golang/protobuf/proto"

	"lark/helper"
	"net"
	"sync"

	"github.com/pkg/errors"
)

type ClientConnInterface interface {
	WriteChan() chan *proto.TaskResp
	Run()
	Close()
}

var JmtextAccess = new(sync.Map)
var ThriftAccess = new(sync.Map)

//接收来自 clientConn 的请求， 打包成 task 之后发送到 TaskReqMsg
type clientConn struct {
	id        string
	parser    *proto.LarkProtoV2
	clientIP  string
	svc       ServerInterface
	writeChan chan *proto.TaskResp
	closeChan chan struct{}
	status    uint32
	protoKind proto.ProtoKind
	conn      net.Conn
}

func NewClient(id string, conn net.Conn, s ServerInterface) ClientConnInterface {
	return &clientConn{
		id:        id,
		status:    newFlag,
		svc:       s,
		writeChan: make(chan *proto.TaskResp, 256),
		closeChan: make(chan struct{}),
		conn:      conn,
	}
}

//接收握手协议，并返回一个版本号
func (c *clientConn) Run() {
	if !atomic.CompareAndSwapUint32(&c.status, newFlag, runningFlag) {
		return
	}

	var buf = make([]byte, 6)
	c.conn.SetReadDeadline(time.Now().Add(conf.DefaultNetReadTimeout()))
	_, err := io.ReadFull(c.conn, buf)
	if err != nil {
		logger.Errorf("[%s] %s", c.id, err)
		c.Close()
		return
	}

	//获取remote ip
	host, _, err := net.SplitHostPort(c.conn.RemoteAddr().String())
	if err != nil {
		logger.Errorf("[%s] error:%s", c.id, err)
		c.Close()
		return
	}
	c.clientIP = host

	if bytes.Contains(buf, []byte("RPC")) {
		host, _, err := net.SplitHostPort(c.conn.RemoteAddr().String())
		if err != nil {
			logger.Errorf("[%s] error:%s", err)
			host = "0.0.0.0"
		}
		JmtextAccess.Store(host, time.Now())

		//jmtext 直连
		c.handleJmtext(buf)
		c.Close()
		return

	} else if bytes.Equal(buf[:2], []byte{27, 88}) && !bytes.Equal(buf[4:6], []byte{128, 1}) {
		// lark client agent 的连接
		err := c.handleLark(buf)
		if err != nil {
			logger.Errorf("[%s] %+v", c.id, err)
			c.Close()
			return
		}

		logger.Debugf("[%s] handshake success.", c.id)

		go c.writeLoop()
		go c.readLoop()

	} else {
		host, _, err := net.SplitHostPort(c.conn.RemoteAddr().String())
		if err != nil {
			logger.Errorf("[%s] error:%s", err)
			host = "0.0.0.0"
		}
		ThriftAccess.Store(host, time.Now())

		c.handleThrift(buf)
		c.Close()
		return
	}
}

func (c *clientConn) handleJmtext(buf []byte) {
	logger.Debugf("[%s] receive jmtext %s", c.id, string(buf))

	br := bufio.NewReader(c.conn)
	line, prefix, err := br.ReadLine()
	if err != nil {
		logger.Errorf("[%s] %s", err)
		return
	}

	if prefix {
		logger.Errorf("line prefix too long")
		return
	}

	l, err := strconv.Atoi(string(line))
	if err != nil {
		logger.Errorf("[%s] %s", err)
		return
	}

	//有的客户端最后有一个 \n 有的没有，此处不读取最后一个 \n ，必须使用短连接
	var data = make([]byte, l+6+len(line)+1)
	copy(data, buf)
	copy(data[6:], line)
	data[6+len(line)] = 10

	_, err = io.ReadFull(br, data[6+len(line)+1:])
	if err != nil {
		logger.Errorf("[%s] %s", err)
		return
	}

	// 清空可能存在的最后一个\n
	c.conn.SetDeadline(time.Now().Add(time.Microsecond * 100))
	br.Read(make([]byte, 1))

	//send task to server
	select {
	case c.svc.TaskChan() <- &TaskReqMsg{Client: c, Req: &proto.TaskReq{Data: data}, Raw: true, CreatedAt: time.Now(), ClientIP: c.clientIP}:
	default:
		logger.Errorf("[%s] send task failed. too many message in queue", c.id)
		c.conn.SetWriteDeadline(time.Now().Add(conf.ServerAgentWriteRemoteTimeout()))
		helper.WriteMcpMsg(c.conn, err)
		return
	}
	logger.Debugf("[%s] send task success", c.id)

	//get resp
	select {
	case resp := <-c.writeChan:
		logger.Debugf("[%s] receive resp from server conn. status:%d, data:%s", c.id, resp.Status, string(resp.Data))
		c.conn.SetWriteDeadline(time.Now().Add(conf.ServerAgentWriteRemoteTimeout()))
		if resp.Status == consts.ServerAgentOK {
			c.conn.Write(resp.Data)
		} else {
			helper.WriteMcpMsg(c.conn, fmt.Errorf("server status:%d", resp.GetStatus()))
		}
		return

	case <-time.After(conf.DefaultNetReadTimeout()):
		err := errors.Errorf("[%s] read task resp timeout", c.id)
		logger.Error(err)

		c.conn.SetWriteDeadline(time.Now().Add(conf.ServerAgentWriteRemoteTimeout()))
		helper.WriteMcpMsg(c.conn, err)
		return
	}

}

func (c *clientConn) handleThrift(buf []byte) {
	logger.Debugf("[%s] receive thrift when handshake", c.id)

	l := binary.BigEndian.Uint32(buf[:4])
	if l > atomic.LoadUint32(&conf.MaxSizeOfNetPackage) {
		var b = make([]byte, 128)
		n, _ := c.conn.Read(b)
		logger.Errorf("[%s] thrift payload too long. %q", b[:n])
		return
	}

	var data = make([]byte, l+4)
	_, err := io.ReadFull(c.conn, data[6:])
	if err != nil {
		logger.Errorf("[%s] %s", c.id, err)
		return
	}

	copy(data[:6], buf)

	select {
	case c.svc.TaskChan() <- &TaskReqMsg{Client: c, Req: &proto.TaskReq{Data: data}, Raw: true, CreatedAt: time.Now(), ClientIP: c.clientIP}:
	default:
		logger.Errorf("[%s] send task failed", c.id)
		return
	}

	select {
	case resp := <-c.writeChan:
		logger.Debugf("[%s] receive resp from server conn. status:%d, data:%q", c.id, resp.Status, resp.Data)
		c.conn.SetWriteDeadline(time.Now().Add(conf.ServerAgentWriteRemoteTimeout()))
		if resp.Status == consts.ServerAgentOK {
			c.conn.Write(resp.Data)
		}

		return

	case <-time.After(conf.LarkClientGetTaskTimeout()):
		logger.Error(err)
		return
	}
}

func (c *clientConn) handleLark(buf []byte) error {
	payload := make([]byte, binary.BigEndian.Uint32(buf[2:6]))
	c.conn.SetReadDeadline(time.Now().Add(conf.DefaultNetReadTimeout()))

	if _, err := io.ReadFull(c.conn, payload); err != nil {
		return errors.WithStack(err)
	}

	var msg = new(proto.Handshake)
	if err := gproto.Unmarshal(proto.SplitPayload(payload)[0], msg); err != nil {
		return errors.WithStack(err)
	}

	//目前只支持一个版本号，只做记录
	logger.Debugf("[%s] receive handshake from client, get version:%d. use version:%d", c.id, msg.GetVersion(), conf.ProtoVersion1)

	var data []byte
	data, err := gproto.Marshal(&proto.Handshake{Version: conf.ProtoVersion1})
	if err != nil {
		return errors.WithStack(err)
	}

	c.parser = proto.NewlarkProtoV2(c.conn, conf.ServerAgentReadRemoteTimeout(), conf.ServerAgentWriteRemoteTimeout())
	err = c.parser.Write(consts.Lsp2lcpTagHandshakeResp, data)
	return errors.WithStack(err)
}

//read data from remote
func (c *clientConn) readLoop() {
	defer util.PrintPanicStack()
	defer c.Close()

	for {
		tag, payload, err := c.parser.ReadTimeout(conf.ServerAgentReadRemoteTimeout())
		if err != nil {
			logger.Errorf("[%s] get message error. %s", c.id, err)
			return
		}

		//TODO check tag
		_ = tag

		var taskr = new(proto.TaskReq)
		if err = gproto.Unmarshal(proto.SplitPayload(payload)[0], taskr); err != nil {
			logger.Errorf("[%s] unmarshal failed. %s", c.id, err)
			return
		}

		logger.Debugf("[%s] receive net data. tag:%d, msgid:%d", c.id, tag, taskr.GetId())

		select {
		case c.svc.TaskChan() <- &TaskReqMsg{Client: c, Req: taskr, CreatedAt: time.Now(), ClientIP: c.clientIP}:
		default:
			logger.Errorf("[%s] business server is too busy")
			var data []byte
			data, err = gproto.Marshal(&proto.TaskResp{
				Id:     taskr.GetId(),
				Status: consts.ServerAgentBusy,
				Data:   taskr.GetData(),
			})
			if err != nil {
				logger.Errorf("[%s] unmarshal failed. %s", err)
				return
			}

			c.parser.WriteWithTimeout(conf.ServerAgentWriteRemoteTimeout(), consts.Lsp2lcpTagTaskResp, data)
		}
	}
}

//write data to remote
func (c *clientConn) writeLoop() {
	defer util.PrintPanicStack()
	defer c.Close()

	for {
		select {
		case resp := <-c.writeChan:
			logger.Debugf("[%s] receive msgid:%d", c.id, resp.GetId())
			b, _ := gproto.Marshal(resp)
			err := c.parser.WriteWithTimeout(conf.ServerAgentWriteRemoteTimeout(), consts.Lsp2lcpTagTaskResp, b)
			if err != nil {
				logger.Errorf("[%s] %s", c.id, err)
				return
			}
			logger.Debugf("[%s] send msgid:%d success", c.id, resp.GetId())

		case <-c.closeChan:
			return
		}
	}
}

func (c *clientConn) closed() bool {
	return atomic.LoadUint32(&c.status) == closedFlag
}

func (c *clientConn) Close() {
	if atomic.CompareAndSwapUint32(&c.status, runningFlag, closedFlag) {
		c.svc.DelClientConn(c.id)
		if c.parser != nil {
			c.parser.Close()
		} else {
			c.conn.Close()
		}
		close(c.closeChan)
	}
}

func (c *clientConn) WriteChan() chan *proto.TaskResp {
	return c.writeChan
}
